package com.yanggu.spark.core.rdd.transform.keyvalue

import org.apache.spark.{SparkConf, SparkContext}

//Key-Value类型-AggregateByKey算子
object RDD19_AggregateByKey {

  def main(args: Array[String]): Unit = {

    //1. 创建sparkConf配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")

    //2. 创建spark上下文对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 从内存中创建RDD
    val rdd = sparkContext.makeRDD(Array(("a", 1), ("b", 2), ("b", 3), ("a", 3), ("b", 4), ("a", 5)), 2)

    //("a", 1), ("b", 2), ("b", 3) => ("a", 1), ("b", 3)
    //("a", 3), ("b", 4), ("a", 5) => ("a", 5), ("b", 4)
    //("a", 6), ("b", 7)

    //4. AggregateByKey
    //按照key进行分组, 将key相同的数据放在一起。形成Iterable
    //aggregateByKey()()使用了函数柯里化
    //函数声明:
    //    def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)] = self.withScope {
    //      aggregateByKey(zeroValue, defaultPartitioner(self))(seqOp, combOp)
    //    }
    //存在两个参数列表
    //1）第一个参数列表表示分区内计算时的初始值（零值）
    //2）第二个参数列表需要传两个参数：
    //  1.第一个参数表示分区内计算规则
    //  2.第二个参数表示分区间计算规则

    //二、需求:创建一个pairRDD，取出每个分区相同key对应值的最大值，然后相加
//    val value = rdd.aggregateByKey(0)((x, y) => math.max(x, y), (x, y) => x + y)
    val value = rdd.aggregateByKey(0)(math.max, _ + _)

    //5. 打印
    value.collect().foreach(println)

    //6. 释放资源
    sparkContext.stop()
  }

}
